package com.xyy.saas.payment.util.mq;

import com.alibaba.fastjson.JSON;
import com.xyy.saas.payment.adpater.RouteManager;
import com.xyy.saas.payment.callback.core.result.CallbackRunner;
import com.xyy.saas.payment.common.Constants;
import com.xyy.saas.payment.common.RocketMqGroup;
import com.xyy.saas.payment.common.RocketMqTopic;
import com.xyy.saas.payment.common.constants.MqConstant;
import com.xyy.saas.payment.cores.bo.QueryResultBo;
import com.xyy.saas.payment.cores.vo.CallbackResultVo;
import com.xyy.saas.payment.dao.model.PayResultOrder;
import com.xyy.saas.payment.dao.model.PrePayOrder;
import com.xyy.saas.payment.service.PayResultOrderService;
import com.xyy.saas.payment.payment.core.service.PrePayOrderService;
import com.xyy.saas.payment.util.ApolloUtil;
import com.xyy.saas.payment.util.JSONUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author: wuqiongda
 * @Date: 2019/12/24 15:38
 */
@Component
@RocketMQMessageListener(topic = RocketMqTopic.TOPIC_QUERY_ORDER_STATE, consumerGroup = RocketMqGroup.GROUP_QUERY_ORDER_STATE ,consumeMode = ConsumeMode.ORDERLY )
public class OrderStateQueryConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
    private static final Logger LOGGER = LoggerFactory.getLogger(OrderStateQueryConsumer.class);
    @Resource
    private RouteManager routeManager;
    @Autowired
    private PayResultOrderService payResultOrderService;
    @Autowired
    private PrePayOrderService prePayOrderService;
    @Autowired
    MqProducer mqProducer;
    @Autowired
    private CallbackRunner callbackRunner;

    @Override
    public void onMessage(MessageExt message) {
        String body = new String(message.getBody());
        Map<String, Object> map = JSONUtils.parse2Map(body, String.class, Object.class);
        try {
            LOGGER.info("OrderStateQueryConsumer onMessage begin topic:{}, body:{}",RocketMqTopic.TOPIC_QUERY_ORDER_STATE, JSON.toJSON(map));
            String type = map.get("type").toString();
            String tradeNo = map.get("tradeNo").toString();
            String businessOrderType = map.get("businessOrderType").toString();
            String payType = map.get("payType").toString();
            int times = Integer.parseInt(map.get("times").toString());
            //构造mq消息
            Map<String, String> mqMsg = new HashMap<>(16);
            mqMsg.put("type", type);
            mqMsg.put("tradeNo", tradeNo);
            mqMsg.put("businessOrderType",businessOrderType);
            mqMsg.put("payType",payType);
            times+=1;
            mqMsg.put("times", String.valueOf(times));
            PrePayOrder prePayOrder = prePayOrderService.getPrePayOrderByPayNo(tradeNo);
            QueryResultBo bo = routeManager.queryPayStatus(tradeNo, type, prePayOrder);
            // 状态改变，更新数据库
            PayResultOrder payResultOrder;
            if (QueryResultBo.STATUS_FINISHED.equals(bo.getStatus())) {
                payResultOrder = payResultOrderService.selectByPayNoAndChannel(tradeNo, Constants.CHANNEL_WX_NAME);
                if (payResultOrder!=null){
                    LOGGER.info("OrderStateQueryConsumer onMessage payResultOrder is exist payNo:{}", tradeNo);
                    return;
                }
                payResultOrder = new PayResultOrder();
                payResultOrder.setCtime(new Date());
                payResultOrder.setUtime(new Date());
                payResultOrder.setReceiveId(prePayOrder.getReceiverId());
                payResultOrder.setAmount(prePayOrder.getAmount());
                payResultOrder.setFee(0);
                payResultOrder.setBusinessOrderType(prePayOrder.getBusinessOrderType());
                payResultOrder.setBusinessType(prePayOrder.getBusinessType());
                payResultOrder.setPayChannel(type);
                payResultOrder.setChannelChargeNo(bo.getChannelChargeNo());
                payResultOrder.setPayNo(prePayOrder.getPayNo());
                int count = payResultOrderService.insertSelective(payResultOrder);
                if (count < 1) {
                    LOGGER.error("OrderStateQueryConsumer onMessage payResultOrderService.insertSelective error payResultOrder:{}", JSON.toJSON(payResultOrder));
                    mqProducer.syncSendDelayMsg(RocketMqTopic.TOPIC_QUERY_ORDER_STATE, JSON.toJSONString(mqMsg), MqConstant.DELAY_LEVELS[Integer.parseInt(ApolloUtil.getProperty("query.order.state.delay.level"))]);
                }
                Long unixTime = System.currentTimeMillis()/1000;
                // 异步回调
                callbackRunner.callback(payResultOrder.getPayNo(), CallbackResultVo.TYPE_CHARGE_SUCCESS, prePayOrder.getAmount(), payResultOrder.getFee(), unixTime, prePayOrder.getCallbackAddress(), type);
            }
            if (QueryResultBo.STATUS_PROCESSING.equals(bo.getStatus())) {
                // 处理重传机制
                int maxTimes = Integer.parseInt(ApolloUtil.getProperty("query.order.state.times"));
                if (times<=maxTimes){
                    mqProducer.syncSendDelayMsg(RocketMqTopic.TOPIC_QUERY_ORDER_STATE, JSON.toJSONString(mqMsg), MqConstant.DELAY_LEVELS[Integer.parseInt(ApolloUtil.getProperty("query.order.state.delay.level"))]);
                    return;
                }
            }
        }catch(Exception e){
            LOGGER.error("OrderStateQueryConsumer onMessage error message:{}", message, e);
        }
        LOGGER.info("OrderStateQueryConsumer onMessage begin topic:{}, body:{}",RocketMqTopic.TOPIC_QUERY_ORDER_STATE, JSON.toJSON(map));
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
    }
}
